package com.yichi.manager;

import android.util.Log;

import com.yichi.global.AppConfig;
import com.yichi.service.DeviceService;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * mqtt服务
 */
public class MqttManager {

    private static final String TAG = "MqttService";

    private MqttClient client;
    private MqttConnectOptions options;
    private IMqttMessageCallback messageCallback;

    public MqttManager() {
        try {
            String clientId = AppConfig.MQTT_CLIENT_ID + AppConfig.DEVICE_ID;
            client = new MqttClient(AppConfig.MQTT_BROKER_URL, clientId, new MemoryPersistence());
            options = new MqttConnectOptions();
            options.setCleanSession(true);//设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
            options.setConnectionTimeout(30);//超时时间，秒
            options.setKeepAliveInterval(30);//心跳，秒
            options.setAutomaticReconnect(true);//自动重连
//            options.setUserName(AppConfig.MQTT_USERNAME);
//            options.setPassword(AppConfig.MQTT_PASSWORD.toCharArray());
            //设置回调
            client.setCallback(mqttCallback);

            Log.v(TAG, "mqtt初始化完成");
        } catch (MqttException e) {
            Log.e(TAG, e.getMessage(), e);
        }
    }

    public void connect() {
        try {
            Log.i(TAG, "mqtt开始连接");
            client.connect(options);//连接不上会阻塞
            Log.i(TAG, "mqtt连接成功");
        } catch (Exception e) {
            Log.e(TAG, e.getMessage(), e);
        }
    }

    //重连
    private void reconnect() {
        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
        scheduler.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                if (!client.isConnected()) {
                    connect();
                }
            }
        }, 0 * 1000, 10 * 1000, TimeUnit.MILLISECONDS);
    }

    public void disconnect() {
        try {
            client.disconnect();
        } catch (MqttException e) {
            Log.e(TAG, e.getMessage(), e);
        }
    }

    public void publishMessage(String topic, String message) {
        if (client == null) return;
        try {
            client.publish(topic, message.getBytes(), 0, false);
        } catch (MqttException e) {
            Log.e(TAG, e.getMessage(), e);
        }
    }

    public void subscribeTopic(String topic) {
        try {
            client.subscribe(topic);
        } catch (MqttException e) {
            Log.e(TAG, e.getMessage(), e);
        }
    }

    public void unsubscribeTopic(String topic) {
        try {
            client.unsubscribe(topic);
        } catch (MqttException e) {
            Log.e(TAG, e.getMessage(), e);
        }
    }

    public void setMessageCallback(IMqttMessageCallback callback) {
        this.messageCallback = callback;
    }

    private MqttCallback mqttCallback = new MqttCallback() {
        @Override
        public void connectionLost(Throwable cause) {
            //失去连接
            Log.i(TAG, "mqtt失去连接...");
            //开始重连
            reconnect();
        }

        @Override
        public void messageArrived(String topic, MqttMessage message) throws Exception {
            Log.i(TAG, "mqtt接收到消息");
            String msg = new String(message.getPayload());

            if (messageCallback != null) {
                messageCallback.handleMessage(msg);
            }

            String log = String.format("mqtt接收到消息：msg:%s;topic:%s;qos:%s;retained:%s", msg, topic, message.getQos(), message.isRetained());
            Log.i(TAG, log);
        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {

        }
    };
}
